package com.example.rabbit.producer.config;

import org.springframework.cloud.function.context.PollableBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import reactor.core.publisher.Flux;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import java.util.function.Supplier;

// 生产者发消息，使用 StreamBridge 手动发的场景居多，
// Spring Cloud Stream 对 Spring Cloud Function 的集成配置注意事项，https://docs.springjava.cn/spring-cloud-stream/reference/spring-cloud-stream/producing-and-consuming-messages.html
@Configuration
public class FunctionConfig {

    @Bean
    public Supplier<String> uppercase() {
        return () -> "Hello from Supplier";
    }
    // 框架提供了一个默认的轮询机制，它将触发对 Supplier 的调用，默认情况下，它将每秒执行一次。换句话说，
    // 上面的配置每秒生成一条消息，每条消息都会发送到绑定器公开的 uppercase-out-0 目标。


    // 但是，想象一下，您想要轮询某个数据源并返回一个表示结果集的有限数据流。响应式编程风格是此类供应商的完美机制。但是，鉴于生成的流的有限性，此类供应商仍然需要定期调用。
    // 该 bean 本身用PollableBean注解（@Bean的子集）进行注解，从而向框架发出信号，尽管此类供应商的实现是响应式的，但它仍然需要轮询。
//    @PollableBean
//    public Supplier<Flux<String>> stringSupplier() {
//        return () -> Flux.just("hello", "bye");
//    }


    // Supplier没有任何输入，因此由不同的机制触发 - 轮询器，它可能具有不可预测的线程机制。
    // 框架还有另一种机制，通过StreamBridge，用户可以更好地控制线程机制。


    // 使用 StreamBridge 的通道拦截器
    // 由于 StreamBridge 使用 MessageChannel 建立输出绑定，因此您可以在通过 StreamBridge 发送数据时激活通道拦截器。应用程序决定对 StreamBridge 应用哪些通道拦截器。
    // Spring Cloud Stream 不会将检测到的所有通道拦截器注入到 StreamBridge 中，除非它们用 @GlobalChannelInterceptor(patterns = "*") 注解。
    @Bean
    @GlobalChannelInterceptor(patterns = "*")
    public ChannelInterceptor customInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                System.out.println("Pre send: " + message);
                return message;
            }
        };
    }

    // 如果您不喜欢上面的全局方法，而希望为每个绑定提供一个专门的拦截器，则可以执行以下操作。
    @Bean
    @GlobalChannelInterceptor(patterns = "foo-*")
    public ChannelInterceptor fooInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                return MessageBuilder.withPayload("hello" + ":1").build();
            }
        };
    }

    @Bean
    @GlobalChannelInterceptor(patterns = "bar-*")
    public ChannelInterceptor barInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                return MessageBuilder.withPayload("hello" + ":1").build();
            }
        };
    }


    // 批处理生产者，参考官方 https://docs.springjava.cn/spring-cloud-stream/reference/spring-cloud-stream/producing-and-consuming-messages.html
    // 您也可以在生产者端使用批处理的概念，方法是返回一个消息集合，这实际上提供了和批处理消费相反的效果，其中集合中的每条消息都将由绑定程序单独发送
    @Bean
    public Function<String, List<Message<String>>> batch() {
        return p -> {
            List<Message<String>> list = new ArrayList<>();
            list.add(MessageBuilder.withPayload(p + ":1").build());
            list.add(MessageBuilder.withPayload(p + ":2").build());
            list.add(MessageBuilder.withPayload(p + ":3").build());
            list.add(MessageBuilder.withPayload(p + ":4").build());
            return list;
        };
    }

}




// 在 Spring Cloud Function 被其他框架（如 Spring Cloud Stream）用作编程模型的典型集成案例中，您需要通过 spring.cloud.function.definition 属性声明要使用哪些函数。
// spring.cloud.function.definition=uppercase,reverse

// 过滤不符合条件的功能
// spring.cloud.function.ineligible-definitions=foo,bar

// Spring Cloud Stream 对 Spring Cloud Function 的集成配置注意事项，https://docs.springjava.cn/spring-cloud-stream/reference/spring-cloud-stream/producing-and-consuming-messages.html
// Spring Cloud Function 文档参考   https://docs.spring.io/spring-cloud-function/reference/index.html
// Spring Cloud Stream 配置参考 https://docs.springjava.cn/spring-cloud-stream/reference/spring-cloud-stream.html
// Function 和 Consumer 的调用触发方式非常简单。它们是根据发送到绑定目标 destination 的数据（事件）触发的。换句话说，它们是典型的事件驱动组件。
// 不过，就触发而言，Supplier 属于自己的类别。顾名思义，它是数据的来源（原点），它不订阅任何绑定目的地 destination，因此必须由其他机制触发。
// 此外，还有一个供应商实施的问题，它可以是命令式的，也可以是反应式的，直接关系到这些供应商的触发问题。

// docker pull rabbitmq
// docker run -d --hostname rabbitmq --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:latest
// docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_management         // 开启管理插件
// docker exec -it rabbitmq rabbitmqctl add_user guest guest                     // 添加用户 guest
// 浏览器访问 http://192.168.56.10:15672/

// 如果出现 All stable feature flags must be enabled after completing an upgrade
//登录 RabbitMQ 的管理界面，导航到 “Admin” -> “Feature Flags”，确保所有稳定的特性标志都是启用状态。如果有任何标志未启用，请将其启用。